package com.tangji.signaltower.disruptor;

import com.fqgj.base.services.redis.RedisClient;
import com.fqgj.common.utils.DateUtil;
import com.jsoniter.output.JsonStream;
import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.WorkHandler;
import com.tangji.signaltower.enums.SendToClientModelEnum;
import com.tangji.signaltower.task.MarketDataVo;
import com.tangji.signaltower.utils.SpringUtils;
import com.tangji.signaltower.bean.MarketDataEntity;
import com.tangji.signaltower.bean.TransBean;
import com.tangji.signaltower.constant.RedisKey;
import com.tangji.signaltower.enums.BizCodeEnum;
import com.tangji.signaltower.service.dao.dao.ProductMarketDao;
import com.tangji.signaltower.service.dao.entity.ProductMarketEntity;
import com.tangji.signaltower.websocket.netty.WebSocketServiceHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Created with IntelliJ IDEA.
 * User: yujianfu (mr.vencnet@gmail.com)
 * Date: 2020-07-03
 * Time: 14:41
 */
public class TradeTransactionInDBHandler implements EventHandler<MarketDataVo>, WorkHandler<MarketDataVo> {

    public static SendToClientModelEnum sendToClientModelEnum = SendToClientModelEnum.NORMAL;

    private static final Logger logger = LoggerFactory.getLogger(TradeTransactionInDBHandler.class);

    private static ThreadPoolExecutor redisExecutor = new ThreadPoolExecutor(1, 1, 5L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(100));
    private static ThreadPoolExecutor mysqlExecutor = new ThreadPoolExecutor(10, 10, 5L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(500));

    private static Map<String, ThreadPoolExecutor> consumerMap = new HashMap<>();

    private static ThreadPoolExecutor getThreadPoolExecutorByProduct(String type) {
        ThreadPoolExecutor executor = consumerMap.get(type);
        if (executor == null) {
            //缓冲区设置为10 如果处理不过来只能丢弃
            executor = new ThreadPoolExecutor(1, 1, 5L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10));
            consumerMap.put(type, executor);
        }
        return executor;
    }

    private RedisClient redisClient;

    private ProductMarketDao productMarketDao;

    @Override
    public void onEvent(MarketDataVo event, long sequence,
                        boolean endOfBatch) throws Exception {
        this.onEvent(event);
    }

    @Override
    public void onEvent(MarketDataVo marketDataEntity) throws Exception {

        if (marketDataEntity == null) {
            logger.error("收到的消息为空");
            return;
        }

        if (redisClient == null) {
            redisClient = (RedisClient) SpringUtils.getBean("redisClient");
        }
        if (productMarketDao == null) {
            productMarketDao = (ProductMarketDao) SpringUtils.getBean("productMarketDaoImpl");
        }

        this.addRedis(marketDataEntity);
        this.sendToClient(marketDataEntity);
        this.addToDataBaseAsync(marketDataEntity);
    }

    public void sendToClient(MarketDataVo entity) {
        ThreadPoolExecutor executor = getThreadPoolExecutorByProduct(entity.getType());
        try {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    entity.setMarketTimestamp(entity.getBusinessTime());
                    String entityStr = JsonStream.serialize(entity);
                    long time1 = System.currentTimeMillis();
                    //消息群发
                    String transDataStr = TransBean.convertTransDataStr(BizCodeEnum.S0001.getBizCode(), null, entityStr);
                    WebSocketServiceHandler.sendAllMessage(
                            entity.getType(), transDataStr, entity, redisClient
                    );

                    long time2 = System.currentTimeMillis();
                    logger.info("数据推送sendToClient22,entity:{},延迟:{}ms,耗时:{},time1:{},time2:{}", entityStr, System.currentTimeMillis() - entity.getMarketTimestamp(), time2 - time1, time1, time2);
                }
            });
        } catch (RejectedExecutionException e) {
            logger.warn("异步推送超过处理能力 sendToClient,type:{},marketTimestamp:{},marketPrice:{},", entity.getType(), entity.getMarketTimestamp(), entity.getMarketPrice(), e);
        } catch (Exception e) {
            logger.error("异步推送失败 sendToClient,type:{},marketTimestamp:{},marketPrice:{},", entity.getType(), entity.getMarketTimestamp(), entity.getMarketPrice(), e);
        }
    }

    public void addRedis(MarketDataVo entity) {
        try {
            redisExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    long time1 = System.currentTimeMillis();
                    String entityStr = JsonStream.serialize(entity);
                    //写入redis
                    redisClient.zAdd(RedisKey.ZSET_BINARY_OPTION_PRODUCT_KEY, Double.valueOf(entity.getMarketTimestamp()), entityStr, entity.getType());
                    redisClient.set(RedisKey.ZSET_BINARY_OPTION_PRODUCT_LAST_KEY, entityStr, 2 * 86400, entity.getType());
                    long time2 = System.currentTimeMillis();
                    logger.info("异步入库,entity:{},延迟:{}ms,耗时:{}", entityStr, System.currentTimeMillis() - entity.getMarketTimestamp(), time2 - time1);
                }
            });
        } catch (RejectedExecutionException e) {
            logger.warn("异步入库超过处理能力,type:{},marketTimestamp:{},marketPrice:{},", entity.getType(), entity.getMarketTimestamp(), entity.getMarketPrice(), e);
        } catch (Exception e) {
            logger.error("异步入redis失败,type:{},marketTimestamp:{},marketPrice:{},", entity.getType(), entity.getMarketTimestamp(), entity.getMarketPrice(), e);
        }
    }

    /**
     * 异步入库
     *
     * @param
     */
    public void addToDataBaseAsync(MarketDataVo marketDataVo) {
        try {
            mysqlExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    /*for (MarketDataEntity entity : marketDataVo.getHistoryData()) {
                        ProductMarketEntity productMarket = new ProductMarketEntity();
                        productMarket.setProductCode(entity.getType());
                        String result1 = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date(entity.getMarketTimestamp() * 1000));
                        productMarket.setMarketDate(new Date(entity.getMarketTimestamp()));
                        productMarket.setMarketTimestamp(entity.getMarketTimestamp());
                        productMarket.setMarketPrice(new BigDecimal(entity.getMarketPrice()));
                        productMarket.setBusinessTime(entity.getBusinessTime());
                        productMarket.setRealType(entity.getRealType());
                        productMarket.setGapData(JsonStream.serialize(entity.getHistoryData()));
                        productMarketDao.insert(productMarket);
                    }*/
                    logger.info("=====异步入库操作。。。。。");
                }
            });
        } catch (RejectedExecutionException e) {
            logger.warn("异步入库超过处理能力", e);
        } catch (Exception e) {
            logger.error("异步入库失败", e);
        }
    }
}